use std::{collections::HashMap, sync::Arc, thread};

use oracle::{pool::Pool, Connection, Row, RowValue};
use tokio::{runtime, sync::Mutex};

use super::user;

pub const SQL_TABLE_PREFIX: &str =
"select * from (
select a.OWNER, a.TABLE_NAME, nvl2(a.IOT_TYPE, 'Y', 'N') as IOT, a.TEMPORARY, decode(a.PARTITIONED, 'YES', 'Y', 'N') as PARTITIONED, decode(nvl(a.COMPRESSION, b.DEF_COMPRESSION), 'ENABLED', 'Y', 'N') as COMPRESSION, nvl2(b.INTERVAL, 'Y', 'N') as PARTITION_INTERVAL, nvl2(c.table_name, 'Y', 'N') as external, f.CONSTRAINT_NAME as PK, d.LOG_GROUP_TYPE SUPPLEMENTAL_LOG, e.INDEX_NAME UK, g.CONSTRAINT_NAME FK, h.TRIGGER_NAME TRIGGERS from dba_tables a 
left join dba_part_tables b on a.owner = b.owner and a.table_name = b.table_name   
left join dba_external_tables c on a.owner = c.owner and a.table_name = c.table_name  
left join (SELECT OWNER, TABLE_NAME, LISTAGG(decode(LOG_GROUP_TYPE, 'ALL COLUMN LOGGING', 'ALL', 'FOREIGN KEY LOGGING','FK','PRIMARY KEY LOGGING','PK','UNIQUE KEY LOGGING','UI','USER LOG GROUP','N'), ',') WITHIN GROUP (ORDER BY LOG_GROUP_TYPE) AS LOG_GROUP_TYPE FROM ALL_LOG_GROUPS GROUP BY OWNER, TABLE_NAME) d on a.owner = d.owner and a.table_name = d.table_name
left join (select TABLE_OWNER as OWNER,TABLE_NAME, LISTAGG(INDEX_NAME, ',') WITHIN GROUP (ORDER BY INDEX_NAME) AS INDEX_NAME from dba_indexes where UNIQUENESS = 'UNIQUE' and STATUS = 'VALID' GROUP BY TABLE_OWNER, TABLE_NAME) e on a.owner = e.owner and a.table_name = e.table_name
left join dba_constraints f on a.owner = f.owner and a.table_name = f.table_name and f.CONSTRAINT_TYPE = 'P'
left join (select OWNER,TABLE_NAME, LISTAGG(CONSTRAINT_NAME, ',') WITHIN GROUP (ORDER BY CONSTRAINT_NAME) AS CONSTRAINT_NAME from dba_constraints where CONSTRAINT_TYPE = 'R' GROUP BY OWNER, TABLE_NAME) g on a.owner = g.owner and a.table_name = g.table_name
left join (select TABLE_OWNER as OWNER, TABLE_NAME, LISTAGG(TRIGGER_NAME, ',') WITHIN GROUP (ORDER BY TRIGGER_NAME) AS TRIGGER_NAME from dba_triggers where STATUS = 'ENABLED' GROUP BY TABLE_OWNER, TABLE_NAME) h on a.owner = h.owner and a.table_name = h.table_name   
) t where";
// 查询类型
pub const SQL_TABLE_DATA_TYPE_PREFIX: &str = "select OWNER, TABLE_NAME, COLUMN_NAME, DATA_TYPE, VIRTUAL_COLUMN, HIDDEN_COLUMN, NULLABLE, 'N' as identity_column from dba_tab_cols where";
pub const SQL_12_TABLE_DATA_TYPE_PREFIX: &str = "select OWNER, TABLE_NAME, COLUMN_NAME, DATA_TYPE, VIRTUAL_COLUMN, HIDDEN_COLUMN, NULLABLE, IDENTITY_COLUMN from dba_tab_cols where";
pub const SQL_MODE: bool = true;

// 支持的类型
pub const TYPE_SUPPORT: [&str; 17] = [
    "VARCHAR2",
    "NVARCHAR2",
    "NUMBER",
    "LONG",
    "DATE",
    "RAW",
    "LONG RAW",
    "ROWID",
    "CHAR",
    "NCHAR",
    "BINARY_FLOAT",
    "BINARY_DOUBLE",
    "CLOB",
    "NCLOB",
    "BLOB",
    "TIMESTAMP",
    "UROWID",
];

pub const PAGE_SIZE: usize = user::PAGE_SIZE;

#[derive(Debug, Default, Clone, RowValue)]
pub struct Table {
    // 所属用户
    pub owner: String,
    // 表名
    pub table_name: String,
    // 唯一索引，除去pk
    pub uk: Option<String>,
    // 唯一索引，可空唯一索引
    #[row_value(with = "setnull")]
    pub nullable_uk: Option<String>,
    // 主键
    pub pk: Option<String>,
    // 外键
    pub fk: Option<String>,
    // 自身触发器
    pub triggers: Option<String>,
    // 关联的触发器
    // pub trigger_refs: Vec<String>,
    // 是否为索引组织表
    #[row_value(with = "parse_bool")]
    pub iot: bool,
    // 是否为临时表
    #[row_value(with = "parse_bool")]
    pub temporary: bool,
    // 是否为分区表
    #[row_value(with = "parse_bool")]
    pub partitioned: bool,
    // 间隔分区
    #[row_value(with = "parse_bool")]
    pub partition_interval: bool,
    // 是否为压缩表: ENABLED, DISABLED, NONE
    #[row_value(with = "parse_bool")]
    pub compression: bool,
    // 是否为外部表
    #[row_value(with = "parse_bool")]
    pub external: bool,
    // 不支持的类型
    #[row_value(with = "setnull")]
    pub data_type_unsupport: Option<String>,
    // 有限支持
    #[row_value(with = "setnull")]
    pub data_type_support_limited: Option<String>,
    // 虚拟列
    #[row_value(with = "setnull")]
    pub virtual_columns: Option<String>,
    // 隐藏列
    #[row_value(with = "setnull")]
    pub hidden_columns: Option<String>,
    // 自增列
    #[row_value(with = "setnull")]
    pub identity_column: Option<String>,
    // 附加日志
    #[row_value(with = "setnull")]
    pub supplemental_log: Option<String>,
}

fn parse_bool(row: &Row, column_name: &str) -> oracle::Result<bool> {
    let value: String = row.get(column_name).unwrap();
    if value == "Y" || value.starts_with("Y") {
        Ok(true)
    } else {
        Ok(false)
    }
}

fn setnull(_row: &Row, _column_name: &str) -> oracle::Result<Option<String>> {
    Ok(None)
}

#[derive(Debug, Default, Clone, RowValue)]
pub struct QueryTableUK {
    // 所属用户
    pub owner: String,
    // 表名
    pub table_name: String,
    // 唯一索引
    pub index_name: String,
}

// OWNER, TABLE_NAME, COLUMN_NAME, DATA_TYPE, COLUMN_ID, VIRTUAL_COLUMN
#[derive(Debug, Default, Clone, RowValue)]
struct TableColumn {
    // 所属用户
    #[allow(dead_code)]
    pub owner: String,
    // 表名
    pub table_name: String,
    // 列名
    pub column_name: String,
    // 列类型
    pub data_type: String,
    // 虚拟列
    #[row_value(with = "parse_bool")]
    pub virtual_column: bool,
    // 隐藏列
    #[row_value(with = "parse_bool")]
    pub hidden_column: bool,
    // 是否可空
    #[row_value(with = "parse_bool")]
    pub nullable: bool,
    // 是否为自增列
    #[row_value(with = "parse_bool")]
    pub identity_column: bool,
}

// impl RowValue for Table {
//     fn get(row: &Row) -> Result<Self> {
//         Ok(Table {
//             owner: row.get("owner").unwrap(),
//             table_name: row.get("table_name").unwrap(),
//             iot: row.get("iot"),
//         })
//     }
// }

fn user_table_count(conn: &Connection, owner: &String) -> usize {
    let sql = format!("select count(*) from dba_tables where owner = '{owner}'");
    let row = conn.query_row_as::<(usize,)>(&sql, &[]).unwrap();
    row.0
}

fn inner_query_tables(
    conn: &Connection,
    owner: &String,
    keys: &Vec<String>,
    build_in_tables: &Vec<String>,
    data_type_support: &Vec<String>,
    data_type_support_limited: &Option<Vec<String>>,
    page: usize,
    page_size: usize,
    show_nullable_uk: bool,
) -> Vec<Table> {
    // table_owner = format!("{})", table_owner.trim_end_matches(","));
    // println!("table_owner: {}", table_owner);
    let mut stmt;
    if SQL_MODE {
        let sql = if keys.len() > 0 {
            format!(
                "{SQL_TABLE_PREFIX} OWNER = '{owner}' and TABLE_NAME in ('{}') order by TABLE_NAME",
                keys.join("','")
            )
        } else {
            let low = (page - 1) * page_size + 1;
            let high = page * page_size;
            format!(
                "SELECT * FROM (SELECT a.*, ROWNUM rn FROM ({SQL_TABLE_PREFIX} OWNER = '{owner}' order by TABLE_NAME) a WHERE ROWNUM <= {high}) WHERE rn >= {low}"
            )
        };
        println!("[{:?}] =>sql_text: {sql}", thread::current().id());

        stmt = conn.statement(&sql).build().unwrap();
    } else {
        let sql = if keys.len() > 0 {
            let placeholders = (2..=keys.len() + 1)
                .map(|i| format!(":{}", i))
                .collect::<Vec<String>>()
                .join(",");
            format!(
                "{} OWNER = :1 and TABLE_NAME in ({}) order by TABLE_NAME",
                SQL_TABLE_PREFIX, placeholders
            )
        } else {
            format!(
                "SELECT * FROM (SELECT a.*, ROWNUM rn FROM ({} OWNER = :1 order by TABLE_NAME) a WHERE ROWNUM <= :2) WHERE rn >= :3", SQL_TABLE_PREFIX
            )
        };
        println!("[{:?}] =>sql_text: {sql}", thread::current().id());

        stmt = conn.statement(&sql).build().unwrap();
        stmt.bind(1, owner).unwrap();

        if keys.len() > 0 {
            for (idx, key) in keys.iter().enumerate() {
                stmt.bind(idx + 2, key).unwrap();
            }
        } else {
            let low = (page - 1) * page_size + 1;
            let high = page * page_size;
            stmt.bind(2, &high).unwrap();
            stmt.bind(3, &low).unwrap();
        }
    }

    // NO-BIND: 10 tables in set (1.1145 sec)

    let mut keys = Vec::new();
    let rows = stmt.query_as::<Table>(&[]).unwrap();

    let mut db_user_tables = Vec::new();
    for row_result in rows {
        let table = match row_result {
            Ok(row_result) => row_result,
            Err(e) => {
                println!("ERR: {e}");
                break;
            }
        };

        // 内置用户.表
        let schema_table = format!("{}.{}", table.owner, table.table_name);
        if build_in_tables.contains(&schema_table) {
            println!("--- Build-in Table: {}", schema_table);
            continue;
        }

        keys.push(table.table_name.clone());
        db_user_tables.push(table);
    }

    inner_query_table_attr(
        &conn,
        &owner,
        &keys,
        &mut db_user_tables,
        data_type_support,
        data_type_support_limited,
        show_nullable_uk
    );

    db_user_tables
}

// 查询表的其他属性
/*
// 唯一索引，除去pk
// 主键
// 外键
// 自身触发器
// 关联的触发器
// 是否为外部表
// 不支持的类型
// 虚拟列
// 附加日志
*/
fn inner_query_table_attr(
    conn: &Connection,
    owner: &String,
    keys: &Vec<String>,
    db_user_tables: &mut Vec<Table>,
    data_type_support: &Vec<String>,
    data_type_support_limited: &Option<Vec<String>>,
    show_nullable_uk: bool
) {
    let db_version = conn.server_version().unwrap().0.major();
    let sql_text = if db_version >= 12 {
        SQL_12_TABLE_DATA_TYPE_PREFIX
    } else {
        SQL_TABLE_DATA_TYPE_PREFIX
    };

    // 查询列类型
    let mut stmt;
    if SQL_MODE {
        let sql = if keys.len() > 0 {
            format!("{sql_text} OWNER = '{owner}' and TABLE_NAME in ('{}') and (VIRTUAL_COLUMN = 'YES' or DATA_TYPE not in ('{}')) and COLUMN_NAME != 'SYS_NC00003$' and COLUMN_NAME != 'SYS_NC00004$'", keys.join("','"), data_type_support.join("','").to_uppercase())
        } else {
            // 11g: XMLType 自动生成 SYS_NC00003$ BLOB
            // 12c+: 默认值 自动生成 SYS_NC00004$  RAW
            // 函数索引 自动生成 SYS_NC00005$
            format!("{sql_text} OWNER = '{owner}' and (VIRTUAL_COLUMN = 'YES' or DATA_TYPE not in ('{}')) and COLUMN_NAME != 'SYS_NC00003$' and COLUMN_NAME != 'SYS_NC00004$'", data_type_support.join("','").to_uppercase())
        };
        println!("[{:?}] =>sql_text: {sql}", thread::current().id());
        stmt = conn.statement(&sql).build().unwrap();
    } else {
        let sql = if keys.len() > 0 {
            let placeholders = (2..=keys.len() + 1)
                .map(|i| format!(":{}", i))
                .collect::<Vec<String>>()
                .join(",");
            format!("{} OWNER = :1 and TABLE_NAME in ({}) and (VIRTUAL_COLUMN = 'YES' or DATA_TYPE not in ('{}')) and COLUMN_NAME != 'SYS_NC00003$' and COLUMN_NAME != 'SYS_NC00004$'", sql_text, placeholders, data_type_support.join("','").to_uppercase())
        } else {
            // 11g: XMLType 自动生成 SYS_NC00003$ BLOB
            // 12c+: 默认值 自动生成 SYS_NC00004$  RAW
            // 函数索引 自动生成 SYS_NC00005$
            format!("{} OWNER = :1 and (VIRTUAL_COLUMN = 'YES' or DATA_TYPE not in ('{}')) and COLUMN_NAME != 'SYS_NC00003$' and COLUMN_NAME != 'SYS_NC00004$'", sql_text, data_type_support.join("','").to_uppercase())
        };
        
        stmt = conn.statement(&sql).build().unwrap();
        stmt.bind(1, owner).unwrap();
        if keys.len() > 0 {
            for (idx, key) in keys.iter().enumerate() {
                stmt.bind(idx + 2, key).unwrap();
            }
        }
    }

    let rows = stmt.query_as::<TableColumn>(&[]).unwrap();

    for row_result in rows {
        let table_column = match row_result {
            Ok(row_result) => row_result,
            Err(e) => {
                println!("ERR: {e}");
                continue;
            }
        };

        for table in db_user_tables.iter_mut() {
            if table.table_name != table_column.table_name {
                continue;
            }
            if table_column.virtual_column {
                // 用逗号分割
                if let Some(vc) = table.virtual_columns.clone() {
                    table.virtual_columns = Some(format!("{}{},", vc, table_column.column_name));
                } else {
                    table.virtual_columns = Some(format!("{},", table_column.column_name));
                }
            }
            if table_column.hidden_column {
                // 用逗号分割
                if let Some(hc) = table.hidden_columns.clone() {
                    table.hidden_columns = Some(format!("{}{},", hc, table_column.hidden_column));
                } else {
                    table.hidden_columns = Some(format!("{},", table_column.hidden_column));
                }
            }
            // 再次匹配过滤
            // 存在虚拟列时会把列类型带回
            if data_type_support.contains(&table_column.data_type) {
                continue;
            }

            // 有限支持
            if let Some(data_type_support_limited) = data_type_support_limited {
                if data_type_support_limited.contains(&table_column.data_type) {
                    if let Some(typ) = table.data_type_support_limited.clone() {
                        table.data_type_support_limited = Some(format!(
                            "{}{}:{},",
                            typ,
                            table_column.column_name,
                            table_column.data_type.clone()
                        ));
                    } else {
                        table.data_type_support_limited = Some(format!(
                            "{}:{},",
                            table_column.column_name,
                            table_column.data_type.clone()
                        ));
                    }
                    continue;
                }
            }

            // 不支持
            if let Some(typ) = table.data_type_unsupport.clone() {
                table.data_type_unsupport = Some(format!(
                    "{}{}:{},",
                    typ,
                    table_column.column_name,
                    table_column.data_type.clone()
                ));
            } else {
                table.data_type_unsupport = Some(format!(
                    "{}:{},",
                    table_column.column_name,
                    table_column.data_type.clone()
                ));
            }

            // 自增列
            if !table_column.identity_column {
                continue;
            }
            if table_column.identity_column && !table_column.nullable {
                table.identity_column = Some(String::from("Y"));
            }
        }
        // }
        
    }

    // 查询可空唯一索引
    let sql = format!("select a.TABLE_OWNER OWNER,a.TABLE_NAME,a.INDEX_NAME from (select TABLE_OWNER,TABLE_NAME,INDEX_NAME,COLUMN_NAME from dba_ind_columns where TABLE_OWNER = '{owner}' and TABLE_NAME in ('{}')) a left join dba_tab_columns b on a.TABLE_OWNER = b.OWNER and a.TABLE_NAME = b.TABLE_NAME and a.COLUMN_NAME = b.COLUMN_NAME where b.NULLABLE = 'Y'", keys.join("','"));
    println!("[{:?}] =>sql_text: {sql}", thread::current().id());

    let rows = conn.query_as::<QueryTableUK>(&sql, &[]).unwrap();
    for row_result in rows {
        let query_table_uk = match row_result {
            Ok(row_result) => row_result,
            Err(e) => {
                println!("ERR: {e}");
                continue;
            }
        };

        for table in db_user_tables.iter_mut() {
            if !(table.owner == query_table_uk.owner
                && table.table_name == query_table_uk.table_name)
            {
                continue;
            }

            if let Some(pk) = table.pk.clone() {
                if pk == query_table_uk.index_name {
                    continue;
                }
            }

            // 可空 唯一索引
            table.nullable_uk = if let Some(nullable_uk) = table.nullable_uk.clone() {
                Some(format!(
                    "{},{},",
                    nullable_uk,
                    query_table_uk.index_name.clone()
                ))
            } else {
                Some(query_table_uk.index_name.clone())
            };

            if !show_nullable_uk {
                if let Some(uk) = table.uk.clone() {
                    table.uk = Some(
                        uk.replace(query_table_uk.index_name.as_str(), "")
                            .replace(",,", ",")
                            .trim_start_matches(",")
                            .trim_end_matches(",")
                            .to_string(),
                    );
                }
            }
        }
    }
}

// 查询表
// build_in_tables: schema.table_name
pub async fn query_tables(
    pool: &Pool,
    user_tables: &HashMap<String, Vec<String>>,
    users: &Vec<String>,
    build_in_tables: &Vec<String>,
    data_type_support: &Vec<String>,
    data_type_support_limited: &Option<Vec<String>>,
    page_size: usize,
    show_nullable_uk: bool,
) -> (HashMap<String, Vec<Table>>, usize) {
    // let mut values: Vec<String> = Vec::new();
    // let mut pos = 1;
    // let mut keys: Vec<String> = Vec::new();
    let mut table_count = 0;

    let mut db_user_all_tables = HashMap::new();

    for user in users.clone() {
        let mut tables = match user_tables.get(&user) {
            Some(tables) => tables.to_vec(),
            None => continue,
        };

        let mut db_user_tables: Vec<Table> = Vec::new();
        let rt = runtime::Builder::new_multi_thread()
            .worker_threads(4)
            .enable_io()
            .enable_time()
            .build()
            .unwrap();

        let mut count = tables.len();
        if count == 1 && tables[0].is_empty() {
            // User ""
            // 查询整个用户的情况，需要统计用户的所有表
            count = user_table_count(&pool.get().unwrap(), &user);
            tables.clear();
        }

        // 启用多线程
        let mut handles = vec![];
        let page_count = count / page_size + if count % page_size == 0 { 0 } else { 1 };
        // println!("page_count: {}", page_count);
        let page = Arc::new(Mutex::new(page_count));
        loop {
            let mut lock = page.lock().await;
            let current_page: usize = lock.to_string().parse().unwrap();
            *lock -= 1;

            let mut cloned_tables = tables.clone();
            tables = cloned_tables.split_off(if cloned_tables.len() > page_size {
                page_size
            } else {
                cloned_tables.len()
            });

            let conn = pool.get().unwrap();
            let cloned_build_in_tables = build_in_tables.clone();
            let cloned_user = user.clone();
            let cloned_data_type_support = data_type_support.clone();
            let cloned_data_type_support_limited = data_type_support_limited.clone();
            let handle = rt.spawn(async move {
                inner_query_tables(
                    &conn,
                    &cloned_user,
                    &cloned_tables,
                    &cloned_build_in_tables,
                    &cloned_data_type_support,
                    &cloned_data_type_support_limited,
                    current_page,
                    page_size,
                    show_nullable_uk
                )
                // query(&conn, &cloned_user_list).await
            });
            handles.push(handle);

            if current_page <= 1 {
                break;
            }
        }

        for handle in handles {
            let mut tables = handle.await.unwrap();
            table_count += tables.len();
            db_user_tables.append(&mut tables);
        }
        rt.shutdown_background();

        db_user_all_tables.insert(user.clone(), db_user_tables);
    }

    // println!("{:?}", db_user_all_tables);
    (db_user_all_tables, table_count)
}
